import akshare as ak
import pandas as pd
from sqlalchemy import create_engine, text

# MySQL 连接配置
engine = create_engine('mysql+pymysql://root:123456@localhost:3306/cn_stock_data')

years = range(2025, 2026)  # 2012 ~ 2022
quarters = ["0331", "0630", "0930", "1231"]
report_dates = [f"{year}{q}" for year in years for q in quarters]

def process_cash_flow(report_date_str: str):
    print(f"开始处理 {report_date_str} ...")
    try:
        # 1. 下载现金流量表数据
        df = ak.stock_xjll_em(date=report_date_str)

        if df.empty:
            print(f"⚠️ {report_date_str} 没有数据，跳过")
            return

        # 2. 删除序号列
        if '序号' in df.columns:
            df = df.drop(columns=['序号'])

        # 3. 数字列转换
        numeric_cols = [
            '净现金流-净现金流', '净现金流-同比增长',
            '经营性现金流-现金流量净额', '经营性现金流-净现金流占比',
            '投资性现金流-现金流量净额', '投资性现金流-净现金流占比',
            '融资性现金流-现金流量净额', '融资性现金流-净现金流占比'
        ]
        for col in numeric_cols:
            if col in df.columns:
                df[col] = pd.to_numeric(df[col], errors='coerce')

        # 4. 公告日期处理
        if '公告日期' in df.columns:
            df['公告日期'] = pd.to_datetime(df['公告日期'], errors='coerce').dt.date

        # 5. 新增 report_date 列
        df['report_date'] = pd.to_datetime(report_date_str, format='%Y%m%d').date()

        # 6. 重命名列与数据库一致
        df = df.rename(columns={
            '股票代码': 'stock_code',
            '股票简称': 'stock_name',
            '净现金流-净现金流': 'net_cash_flow',
            '净现金流-同比增长': 'net_cash_flow_yoy',
            '经营性现金流-现金流量净额': 'operating_cash_flow',
            '经营性现金流-净现金流占比': 'operating_cash_flow_ratio',
            '投资性现金流-现金流量净额': 'investing_cash_flow',
            '投资性现金流-净现金流占比': 'investing_cash_flow_ratio',
            '融资性现金流-现金流量净额': 'financing_cash_flow',
            '融资性现金流-净现金流占比': 'financing_cash_flow_ratio',
            '公告日期': 'announcement_date'
        })

        # 7. 删除旧数据（避免重复）
        with engine.begin() as conn:
            conn.execute(
                text("DELETE FROM cash_flow_statement WHERE report_date = :report_date"),
                {"report_date": df['report_date'].iloc[0]}
            )

        # 8. 插入 MySQL
        df.to_sql('cash_flow_statement', con=engine, if_exists='append', index=False)

        print(f"✅ {report_date_str} 导入完成！ 共 {len(df)} 行")

    except Exception as e:
        print(f"❌ {report_date_str} 处理失败: {e}")

# 批量处理
for date in report_dates:
    process_cash_flow(date)

print("🎉 2015-2025 所有季度现金流量表导入完成！")
